package com.usian.admin.config;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.usian.admin.controller.AdNewsController;
import com.usian.admin.fegin.ArtivleFegin;
import com.usian.common.medis.NewsAutoScanConstants;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Optional;

@Component
public class WemediaNewsAutoListener {

    @Autowired
    private AdNewsController adNewsController;

    @Autowired
    private ArtivleFegin artivleFegin;

    //kakfa监听的主题
    @KafkaListener(topics = NewsAutoScanConstants.Wm_NEWS_AUTO_SCAN_TOPIC)
    public void recevieMessage(ConsumerRecord<?,?> record) throws Exception {
        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record);
        if(optional.isPresent()){
            Object value = record.value();
            //调用自动审核的方法
            adNewsController.findByIdNews(Integer.valueOf((String) value) );
        }
    }

    @KafkaListener(topics = NewsAutoScanConstants.Wm_News_enable)
    public void newsEnable(ConsumerRecord<?,?> record) throws Exception {
        Optional<? extends ConsumerRecord<?, ?>> optional = Optional.ofNullable(record);
        if(optional.isPresent()){
            Object value = record.value();
            //调用自动审核的方法
            String s = value.toString();
            Map<String, String> map = JSONObject.parseObject(s, new TypeReference<Map<String, String>>() {});
            String articel = map.get("Articel");
            String enable = map.get("enable");
            artivleFegin.UpdateArticle_Config(enable,articel);
            //adNewsController.NewsEnableUpdateArticle(s);
        }
    }
}
